package mapreduce;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;

public class Record {
	public static class RecordData {
		public RecordData(Object data, BytesFormat format) {
			m_data = data;
			m_format = format;
		}

		public Object getData() {
			return m_data;
		}

		public BytesFormat getFormat() {
			return m_format;
		}

		public boolean equals(Object anotherObject) {
			if (getClass() != anotherObject.getClass()) {
				return false;
			}
			RecordData anotherData = (RecordData) anotherObject;
			if (m_data.equals(anotherData.getData())
					&& m_format.equals(anotherData.getFormat())) {
				return true;
			} else {
				return false;
			}
		}

		private Object m_data;
		private BytesFormat m_format;
	}

	public static class RecordPositionType {
		RecordData getRecordData(Record record) {
			if (this.equals(getKeyType())) {
				return record.getKey();
			} else if (this.equals(getValueType())) {
				return record.getValue();
			}
			return null;
		}

		public static RecordPositionType getKeyType() {
			return new RecordPositionType(1);
		}

		public static RecordPositionType getValueType() {
			return new RecordPositionType(2);
		}

		private RecordPositionType(int code) {
			m_code = code;
		}

		public boolean equals(Object anotherObject) {
			if (getClass() != anotherObject.getClass()) {
				return false;
			}
			RecordPositionType anotherType = (RecordPositionType) anotherObject;
			if (m_code == anotherType.m_code) {
				return true;
			} else {
				return false;
			}
		}

		int m_code;
	}

	public static class RecordPosition {
		public RecordPosition(RecordPositionType type, int position) {
			m_type = type;
			m_position = position;
		}

		public RecordPositionType getType() {
			return m_type;
		}

		public int getPosition() {
			return m_position;
		}

		RecordPositionType m_type;
		int m_position;
	}

	static public Record getRecord(RecordData key, RecordData value) {
		return new Record(key, value);
	}

	static private RecordData getRecordData(InputStream in) throws IOException {
		byte[] lengthBytes = new byte[4];
		in.read(lengthBytes);
		int length = ByteBuffer.wrap(lengthBytes).getInt();
		if (length == 0) {
			throw new IOException();
		}

		byte[] bytes = new byte[length];
		in.read(bytes);
		return BytesFormat.decodeBuffer(bytes);
	}

	static public Record getRecord(InputStream in) throws IOException {
		RecordData keyData = getRecordData(in);
		RecordData valueData = getRecordData(in);
		return getRecord(keyData, valueData);
	}

	public RecordData getKey() {
		return m_key;
	}

	public RecordData getValue() {
		return m_value;
	}

	private byte[] getBytes(RecordData recordData) {
		byte[] bytes = BytesFormat.encodeBuffer(recordData);

		ByteBuffer buffer = ByteBuffer.allocate(bytes.length + 4);
		buffer.putInt(bytes.length);
		buffer.put(bytes);
		byte[] result = new byte[bytes.length + 4];
		buffer.rewind();
		buffer.get(result);
		return result;
	}

	public byte[] getBytes() {
		if (m_keyBytes == null || m_valueBytes == null) {
			m_keyBytes = getBytes(m_key);
			m_valueBytes = getBytes(m_value);
		}
		byte[] bytes = new byte[m_keyBytes.length + m_valueBytes.length];
		int index = 0;
		for (int i = 0; i < m_keyBytes.length; i++) {
			bytes[index] = m_keyBytes[i];
			index++;
		}
		for (int i = 0; i < m_valueBytes.length; i++) {
			bytes[index] = m_valueBytes[i];
			index++;
		}
		return bytes;
	}

	public byte[] getReadableBytes() {
		byte[] keyBytes = BytesFormat.formatBuffer(m_key);
		byte[] valueBytes = BytesFormat.formatBuffer(m_value);
		byte[] bytes = new byte[keyBytes.length + valueBytes.length + 1];
		
		int index = 0;
		for(int i = 0 ; i < keyBytes.length ; i ++){
			bytes[index] = keyBytes[i];
			index ++;
		}
		bytes[index] = '\n';
		index ++;
		for(int i = 0 ; i < valueBytes.length ; i ++){
			bytes[index] = valueBytes[i];
			index ++;
		}
		return bytes;
	}

	private Record(RecordData key, RecordData value) {
		m_key = key;
		m_value = value;
	}

	public BytesFormat getKeyFormat() {
		return m_key.getFormat();
	}

	public BytesFormat getValueFormat() {
		return m_value.getFormat();
	}

	public Object getEntry(RecordPosition position) {
		RecordData data = position.getType().getRecordData(this);
		if (data.getData().getClass() != Object[].class) {
			return data.getData();
		} else {
			return ((Object[]) data.getData())[position.getPosition()];
		}
	}

	public BytesFormat getFormat(RecordPosition position) {
		RecordData data = position.getType().getRecordData(this);
		if (data.getFormat().getClass() != MultipleTypeFormat.class) {
			return data.getFormat();
		} else {
			return ((MultipleTypeFormat) data.getFormat())
					.getSubFormat(position.getPosition());
		}
	}

	byte[] getKeyBytes() {
		if (m_keyBytes == null) {
			getBytes();
		}
		return m_keyBytes;
	}

	byte[] getValueBytes() {
		if (m_valueBytes == null) {
			getBytes();
		}
		return m_valueBytes;
	}

	public boolean equals(Object anotherObject) {
		if (getClass() != anotherObject.getClass()) {
			return false;
		}
		Record anotherRecord = (Record) anotherObject;
		byte[] thisBytes = getBytes();
		byte[] anotherBytes = anotherRecord.getBytes();
		if (thisBytes.length != anotherBytes.length) {
			return false;
		}
		for (int i = 0; i < thisBytes.length; i++) {
			if (thisBytes[i] != anotherBytes[i]) {
				return false;
			}
		}
		return true;
	}

	RecordData m_key = null;
	RecordData m_value = null;
	byte[] m_keyBytes = null;
	byte[] m_valueBytes = null;

	static int BUFFER_SIZE = 16 * 1024;
}
